package com.atguigu.day01;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.*;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

public class Flink01_Batch_WordCount {
    public static void main(String[] args) throws Exception {
        //1.创建批的执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        //2.读取文本数据
        DataSource<String> dataSource = env.readTextFile("input/word.txt");

       /*Spark:（
       flatMap（将一行数据切出每一个单词）
       ->map(将单词组成Tuple2元组)
       ->reduceBykey(
                     1.将相同的单词聚合到一块（bykey）
                     2.对value相加（redue）)
       ->print打印到控制台）*/

       //3.将一行数据切出每一个单词
        FlatMapOperator<String, String> wordData = dataSource.flatMap(new MyFlatMap());

        //4.将单词组成Tuple2元组
        MapOperator<String, Tuple2<String, Integer>> wordToOneData = wordData.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
//                return new Tuple2<>(value, 1);
                return Tuple2.of(value, 1);
            }
        });
        //5.将相同的单词聚合到一块
        UnsortedGrouping<Tuple2<String, Integer>> groupBy = wordToOneData.groupBy(0);

        //6.对value进行相加
        AggregateOperator<Tuple2<String, Integer>> sum = groupBy.sum(1);

        //7.打印到控制台
        sum.print();
    }

    public static class MyFlatMap implements FlatMapFunction<String,String>{

        @Override
        public void flatMap(String value, Collector<String> out) throws Exception {
            //将一行数据按照空格切分
            String[] words = value.split(" ");
            //遍历数据获取每一个单词
            for (String word : words) {
                //通过采集器将数据发送至下游
                out.collect(word);
            }
        }
    }
}
